package com.xxx.flink.transform;

import com.xxx.flink.customsource.CustomSource;
import com.xxx.flink.pojo.Event;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 简单聚合
 */
public class TransformSimpleAggTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取数据
        DataStreamSource<Event> sourceStream = env.addSource(new CustomSource());

        // 分组聚合：获取每个用户最近一次的访问时间
        sourceStream.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event event) throws Exception {
                return event.getName();  // 使用name分组
            }
        }).max("timestamp").print("max:");   // 使用max()获取timestamp字段最大值

        // Lambda表达式写法
        sourceStream.keyBy(Event::getName)
                .maxBy("timestamp").print("maxBy:");   // maxBy()


        env.execute();
    }
}
